package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.pojo.vo.ChargingProcessVO;
import cn.tedu.charging.common.utils.JsonUtils;
import cn.tedu.charging.common.web.JsonResult;
import cn.tedu.charging.order.constant.Constant;
import cn.tedu.charging.order.constant.MqttConstant;
import cn.tedu.charging.order.constant.OrderStatusConstant;
import cn.tedu.charging.order.dao.repository.OrderRepository;
import cn.tedu.charging.order.feign.CostClient;
import cn.tedu.charging.order.feign.UserClient;
import cn.tedu.charging.order.pojo.param.ChargingResultParam;
import cn.tedu.charging.common.pojo.param.ChargingProcessParam;
import cn.tedu.charging.order.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.order.pojo.po.ChargingProcessPO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

import java.math.BigDecimal;

/**
 * 用来消费设备发送的消息
 */
@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    public MqttConsumer(MqttClient mqttClient,
                        MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    private MqttClient mqttClient;

    private MqttContext mqttContext;

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.debug("MqttConsumer:连接成功");
        try {
            //和EMQ连接成功后,订阅/topic/start/result 来消费充电桩发送 充电是否成功 的指令
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_RESULT, 0);
            //EMQ连接成功后,订阅/topic/charging/process 来消费充电桩发送 充电数据 的指令
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_PROCESS, 0);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }

    }

    @Override
    public void connectionLost(Throwable cause) {
        log.debug("MqttConsumer:连接失败", cause);
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("MqttConsumer:消息到达:topic-{},message-{}", topic, message);
        //设备同步 是否开始充电
        if (MqttConstant.TOPIC_CHARGING_RESULT.equals(topic)) {
            handleChargingResult(message);
            //设备同步 充电进度
        }else if (MqttConstant.TOPIC_CHARGING_PROCESS.equals(topic)) {
            handleChargingProcess(message);
        }
    }

    /**
     * 处理 设备同步的充电状态 充电进度 数据
     * @param message
     */
    private void handleChargingProcess(MqttMessage message){
        log.debug("MqttConsumer:消息到达 handleChargingProcess:message-{}",message);
        ChargingProcessParam chargingProcessParam = JsonUtils.fromJson(message.toString(), ChargingProcessParam.class);

        //1 计算电费  加另外一个微服务 计价服务
        ChargingProcessVO orderCostResult = getOrderCost(chargingProcessParam);
        if (orderCostResult != null) {
            //通过计价服务获取的当前充电进度的总花费
            BigDecimal totalCost = orderCostResult.getTotalCost();

            //2 判断温度 大于45度 给用户推送(Websocket)B级告警 大于50 推送 A级告警
            float temperature = chargingProcessParam.getTemperature();
            if (temperature > MqttConstant.A_WARNING) { //50 应该定义为常量
                log.debug("温度超过:{} 给用户发送A级告警通知,直接停止设备,实际温度:{}",MqttConstant.A_WARNING,temperature);
            }else if (temperature > MqttConstant.B_WARNING) {
                log.debug("温度超过:{} 给用户发送B级告警通知,设备不停止,让用户选择,实际温度:{}",MqttConstant.B_WARNING,temperature);
            }
            //3 判断电费是否大于余额
            //获取当前用户余额
            BigDecimal userBalance = getUserBalance(chargingProcessParam.getUserId());
            //用户的余额减去电费如果 小于 0
            if((userBalance.subtract(totalCost)).floatValue() <= 0 ){
                log.debug("给用户发通知 余额不足,停止充电,告诉用户需要充值.当前余额:{},订单消费:{}",userBalance,totalCost);
            }
            //4 判断是否充满 todo
            //5 推送(Websocket)充电进度到手机端
            syncChargingProcess2User(orderCostResult);
            //6 充电进度保存 时序数据的保存
            saveChargingProcess(chargingProcessParam);
        }


    }

    /**
     * 保存充电进度数据
     * @param chargingProcessParam
     */
    private void saveChargingProcess(ChargingProcessParam chargingProcessParam) {
        log.debug("保存充电进度数据:{}",chargingProcessParam);
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        //时间:value
        //时间
        chargingProcessPO.setId(System.currentTimeMillis());
        //value
        BeanUtils.copyProperties(chargingProcessParam,chargingProcessPO);
        mqttContext.getChargingProcessRepository().save(chargingProcessPO);
        log.debug("保存充电进度数据成功:{}",chargingProcessParam);
    }

    /**
     * 给用户同步充电数据同步到手机端
     * @param chargingProcessVO
     */
    private void syncChargingProcess2User(ChargingProcessVO chargingProcessVO){
        log.debug("给用户同步充电数据同步到手机端:{}",chargingProcessVO);
    }


    /**
     * 获取用户的余额
     * @return
     */
    private BigDecimal getUserBalance(Integer userId) {
        log.debug("调用用户服务获取用户余额:{}",userId);
        JsonResult<BigDecimal> userBalance = mqttContext.getUserClient().getUserBalance(userId);
        log.debug("调用用户服务获取用户余额:入参:{},出参:{}",userId,userBalance);
        if (userBalance != null) {
            return userBalance.getData();
        }
        return new BigDecimal(0);
    }

    /**
     * 获取电费
     * @return
     */
    private ChargingProcessVO getOrderCost(ChargingProcessParam chargingProcessParam) {
        JsonResult<ChargingProcessVO> result
                = mqttContext.getCostClient().calculateCost(chargingProcessParam);
        if (result != null) {
            ChargingProcessVO data = result.getData();
            if (data != null) {
                return data;
            }
        }
        return null;
    }

    /**
     * 处理 设备同步的充电结果
     * @param message
     */
    private void handleChargingResult(MqttMessage message) {
        log.debug("MqttConsumer:消息到达 handleChargingResult:message-{}",message);
        ChargingResultParam deviceResultParam = JsonUtils.fromJson(message.toString(), ChargingResultParam.class);
        if (deviceResultParam != null) {
            String result = deviceResultParam.getResult();
            if (Constant.RESULT_START_CHARGING_SUCCESS.equals(result)) {
                log.debug("充电桩开始充电:{}", deviceResultParam);
                //创建订单成功记录
                //orderService.saveSuccessOrder(deviceResultParam);
                ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();
                chargingBillSuccessPO.setBillId(deviceResultParam.getOrderNo());
                //订单状态为充电中
                chargingBillSuccessPO.setBillStatus(OrderStatusConstant.ORDER_STATUS_PROGRESS);
                log.debug("保存订单数据:{}", chargingBillSuccessPO);
                //如果设备重复发送消息,
                //会导致订单成功表  charging_bill_success 会重复的保存订单数据
                // 1 在订单表 给 订单号创建唯一索引 不能重复 重复插入数据
                //Duplicate entry '10000_80_1719887147282' for key 'charging_bill_success_bill_id_unique_key'
                // 异常可以处理,可以不处理 通常是接入到监控系统
                // 2 先查询,如果没有,插入,如果有 跳过 代码怎么写的问题 原子性的问题

                //需要记录状态,如果处理过了,就不再重复处理, redis的布隆过滤器 自己了解

                Integer row = mqttContext.getOrderRepository().saveSuccessOrder(chargingBillSuccessPO);
                log.debug("保存订单数据成功:{},返回结果:{}", chargingBillSuccessPO, row);
            } else {
                log.debug("充电桩故障");
                //创建订单失败记录
                ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
                chargingBillFailPO.setBillId(deviceResultParam.getOrderNo());
                chargingBillFailPO.setFailDesc("设备自检失败-设备有响应但是充不了电");
                mqttContext.getOrderRepository().saveFailOrder(chargingBillFailPO);
            }
        }
    }


    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("MqttConsumer deliveryComplete");
    }
}
